package com.aotain.nyx.udpflood;

import java.util.HashMap;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import com.aotain.nyx.flink.BasicTuple;

/**
 * 
     *@Title:UDP Flood 检测 MAP
     *@Description:
     *@Author:Turk
     *@Since:2017年8月30日
     *@Version:1.1.0
 */
public class UDPFloodMap implements MapFunction<Tuple2<String,String>, Tuple2<String,HashMap<String,BasicTuple>>>{

	/**
	 *
	 */
	    
	private static final long serialVersionUID = 3679823857255749074L;

	@Override
	public Tuple2<String, HashMap<String,BasicTuple>> map(Tuple2<String, String> value)
			throws Exception {
		
//		// [0-4]dip+ "|" +dport+ "|" +sip+ "|" +sport+ "|" +gisstrBase64+ "|" + 
		// [5-9]domainname+ "|" +url+ "|" +type+ "|" +delaytime+ "|" +idcno+ "|" + 
		// [10-14]upStreamOctets+ "|" +upStreamPacket+ "|" +status+ "|" +userAgent+ "|" +cookie+ "|" + 
		// [15-19]dnStreamOctets+ "|" +dnStreamPacket+ "|" + refer + "|" +post | OUT/IN
		
		Tuple2<String, HashMap<String,BasicTuple>> ret = new Tuple2<String, HashMap<String,BasicTuple>>();
		String logString = value.f1; //格式化后的原始数据
		String[] items = logString.split("\\|",-1);
		if(items.length <= 10)
			return ret;
		
		String protocaltype = items[7];
		
		if(!protocaltype.equals("17")) //判断是否UDP协议
			return ret;
		
		String groupip = value.f0;
		String sip = items[2];
		String gis = items[4];
		String idc = items[9];
		String dip = items[0];
		String destport = items[1];
		String flowdirection = items[19];
		
		long upStreamOctets = 0; //UPSTREAMOCTETS
		long upStreamPacket = 0; //UPSTREAMPACKET
		
		String tripleKey = dip + "_" + sip + "_" + destport  +"_" + gis + "_" + idc;
		
		upStreamOctets = Long.parseLong(items[10]);
		upStreamPacket = Long.parseLong(items[11]);
		
		HashMap<String,BasicTuple> mapTuple = new HashMap<String,BasicTuple>();
		
		BasicTuple tuple = new BasicTuple();
		tuple.setDestIP(dip);
		tuple.setSourceIP(sip);
		tuple.setDestPort(destport);
		tuple.setGIS(gis);
		tuple.setIDC(idc);
		tuple.setUpStreamOctets(upStreamOctets);
		tuple.setUpStreamPacket(upStreamPacket);
		tuple.setFlowDirection(flowdirection);
		
		mapTuple.put(tripleKey, tuple);
		
		ret.setFields(groupip + "_" + destport, mapTuple);
		
		return ret;
		    
	}

}

    